/**
 * Illustrates joining two csv files
 */
package com.hsj;

import java.io.StringReader;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import au.com.bytecode.opencsv.CSVReader;

import org.apache.commons.lang.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;

public class BasicLoadWholeCsv {
    /**
     * 配合wholeTextFiles方法使用 用来读取文件夹下的所有文件
     */
    public static class ParseLine implements FlatMapFunction<Tuple2<String, String>, String[]> {
        public Iterator<String[]> call(Tuple2<String, String> file) throws Exception {
//            System.out.println(file._1+"--"+file._2);
            CSVReader reader = new CSVReader(new StringReader(file._2()));
            return reader.readAll().iterator();
        }
    }

    public static void main(String[] args) throws Exception {
//        if (args.length != 3) {
//            throw new Exception("Usage BasicLoadCsv sparkMaster csvInputFile csvOutputFile key");
//        }
        String master = "local";//args[0];
        String csvInput = "files/favourite_animals.csv";//args[1];
        String outputFile = "BasicLoadWholeCsv";//args[2];
        final String key = "spark";//args[3];

        JavaSparkContext sc = new JavaSparkContext(
                master, "loadwholecsv", System.getenv("SPARK_HOME"), System.getenv("JARS"));
        JavaPairRDD<String, String> csvData = sc.wholeTextFiles(csvInput);
        JavaRDD<String[]> keyedRDD = csvData.flatMap(new ParseLine());

        JavaRDD<String[]> result =
                keyedRDD.filter(new Function<String[], Boolean>() {
                    public Boolean call(String[] input) {
                        return input[0].equals(key);
                    }
                });
        result.foreach(new VoidFunction<String[]>() {
            @Override
            public void call(String[] strings) throws Exception {
                StringBuffer sb = new StringBuffer();
                for (String str:strings) {
                    sb.append(str).append(",");
                }
                System.out.println(sb.toString());
            }
        });
//        result.saveAsTextFile(outputFile);
    }
}
